/*******************************************************************************
 * Copyright (c) 2005, 2014 springside.github.io
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 *******************************************************************************/
package com.producer;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;

import org.springframework.boot.actuate.health.Health;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import com.PiwikPredictionioSysCfg;
import com.PiwikPredictionioSysCfgDao;

@Component
public class PiwikLogVisitDataScanner extends DataScanner {
	public static final String topic = "piwik.piwik_log_visit";
	@Override
	public Health health() {
		return Health.up().withDetail("piwik_log_visit Scanner", this.state()).build();
	}

	@Override
	@Transactional
	protected void exec() throws Exception {
//		System.err.println(getClass().getName()+":"+kafkaProducerService);
		PiwikPredictionioSysCfg sysCfg = 
						piwikPredictionioSysCfgDao.
				findOne(PiwikPredictionioSysCfgDao.key_lastInsertedPiwikLogVisitId);
		
		Integer lastInsertedId = new Integer(sysCfg.getValue());	
		List<Map<String, Object>>  list = piwikdao.queryPiwikLogVisitList(lastInsertedId);
		if (list.isEmpty()) {
			return ;
		}
		String value = "";
		for (Map<String, Object> data : list) {
			String id = data.getOrDefault("idvisit", "").toString();
			try {
				kafkaProducerService.sendMessgae(topic, id, data);
			} catch (InterruptedException | ExecutionException e) {
				break;
			}
			value=id;
		};
		
		sysCfg.setValue(value);
		piwikPredictionioSysCfgDao.save(sysCfg);
	}


		




}
